package org.ricks.gateway.cluster.transport;

import io.scalecube.cluster.transport.api.Message;
import lombok.Setter;
import org.ricks.ioc.utils.Logger;
import org.ricks.net.StateMachineEnum;
import org.ricks.net.extension.processor.AbstractMessageProcessor;
import org.ricks.net.transport.AioSession;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;

// 消息处理器
@Setter
public class SmartSocketMessageProcessor extends AbstractMessageProcessor<Message> {

    private Function<Message, CompletableFuture<Void>> listener;

    private CompletableFuture<AioSession> sessionFuture;

    @Override
    public void process0(AioSession session, Message message) {
        if (listener != null) {
            listener.apply(message).exceptionally(ex -> {
                Logger.error("Message processing error", ex);
                return null;
            });
        }
    }

    @Override
    public void stateEvent0(AioSession session, StateMachineEnum state, Throwable error) {
        switch (state) {
            case NEW_SESSION:
                logSessionEvent(session, "Created");
                sessionFuture.complete(session);
                break;
            case PROCESS_EXCEPTION:
                Logger.error("Processing error: {}", error,error.getMessage());
                session.close();
                break;
            case DECODE_EXCEPTION:
                Logger.error("Decoding error", error);
                break;
            case SESSION_CLOSED:
                logSessionEvent(session, "Closed");
                break;
        }
    }

    private void logSessionEvent(AioSession session, String action) {
        try {
            Logger.info("Session {}: {} -> {}",
                    action, session.getLocalAddress(), session.getRemoteAddress());
        } catch (IOException e) {
            Logger.warn("Address resolution error", e);
        }
    }
}